package apriorialgo;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class CKItemsetMapper extends
		Mapper<LongWritable, Text, Text, IntWritable> {
	private ArrayList<ArrayList<String>> candidateItemsets = new ArrayList<ArrayList<String>>();
	private ArrayList<ArrayList<String>> kMinus1Itemsets = new ArrayList<ArrayList<String>>();
	private int kMinus1 = 2;  // size of L itemset or k - 1
	
	/**
	 * Reads the L(k-1) frequent itemsets
	 */
	@Override
	protected void setup(Context context) throws IOException,
			InterruptedException {
		super.setup(context);		
		Configuration conf = context.getConfiguration();
		kMinus1 = conf.getInt("itemset.k", 1) - 1;   
		System.out.println("Enter CKItemsetMapper.setup k = " + (kMinus1 + 1));
		try {
			Path[] cacheFiles = DistributedCache.getLocalCacheFiles(conf);
			if (cacheFiles != null && cacheFiles.length > 1) {
				System.out.println("cacheFiles size: " + cacheFiles.length);
				for (Path p : cacheFiles) {
					System.out.println("Cache file: " + p.toUri().toString());
				}
				System.out.println("reading cache file: " + cacheFiles[1].toUri().toString());
				BufferedReader bufReader = new BufferedReader(new FileReader(
						cacheFiles[1].toString())); // the (k-1)-items frequent itemsets file
				try {					
					String line;
					while ((line = bufReader.readLine()) != null) {
						StringTokenizer tokenizer = new StringTokenizer(line);
						ArrayList<String> frequentItemset = new ArrayList<String>();
						for (int i = 0; i < kMinus1; i++) {
							frequentItemset.add(tokenizer.nextToken().trim());
						}
						kMinus1Itemsets.add(frequentItemset);
					}
				} finally {
					bufReader.close();
				}
			}
		} catch (IOException e) {
			System.err.println("Exception reading DistributedCache: " + e);
		}
		System.out.println("Exit.  no. frequent itemset of size " + kMinus1 + ": " +
				+ kMinus1Itemsets.size());
	}

	/**
	 * Check if the specified freqItemset subset is also frequent
	 * @param freqItemset the frequent itemset to check
	 * @return True if freqItemset is found in the k-1 frequent itemsets, False otherwise
	 */
	private boolean isFrequentItemset(ArrayList<String> freqItemset) {		
		Iterator<ArrayList<String>> itemsetIter = kMinus1Itemsets.iterator();
		while (itemsetIter.hasNext()){
			ArrayList<String> itemset = itemsetIter.next();
			boolean frequent = false;
			for (int i = 0; i < freqItemset.size(); i++) {
				if (itemset.contains(freqItemset.get(i))) {
					frequent = true;					
				}
				else {
					frequent = false;
					break;
				}
			}
			if (frequent) {
				// found it
				return true;
			}
		}
		return false;
	}
	
	/**
	 * Generates candidate frequent itemsets
	 */
	@Override
	public void map(LongWritable key, Text value, Context context)
			throws IOException {
		System.out.println("Enter CKItemsetMapper.map k = " + (kMinus1 + 1));
		// Join the k-1 frequent itemset to itself		
		ArrayList<ArrayList<String>> kItemsets = new ArrayList<ArrayList<String>>();		
		ArrayList<String> candidate = new ArrayList<String>();
		StringTokenizer tokenizer = new StringTokenizer(value.toString());
		for (int i = 0; i < kMinus1; i++) {
			candidate.add(tokenizer.nextToken().trim());
		}
		Iterator<ArrayList<String>> itemsetIter = kMinus1Itemsets.iterator();
		while (itemsetIter.hasNext()){
			ArrayList<String> itemset = itemsetIter.next();
			Iterator<String> itemIter = itemset.iterator();
			while (itemIter.hasNext()) {
				String item = itemIter.next();
				if (!candidate.contains(item)) {
					ArrayList<String> jointCandidate = (ArrayList<String>)candidate.clone();
					jointCandidate.add(item);
					// sort the new candidate which will be used as keys to be reduced
					Collections.sort(jointCandidate, new Comparator<String>() {
						// Compare numbers if input converts to number
						// Otherwise compare alphanumerically
						@Override
						public int compare(String object1, String object2) {
							// TODO Auto-generated method stub
							int val1, val2;
							try {
								val1 = Integer.parseInt(object1);
								val2 = Integer.parseInt(object2);
								if (val1 == val2) {
									return 0;
								}
								else {
									return (val1 > val2 ? 1 : -1);
								}
							}
							catch (NumberFormatException ex) {
								return object1.compareTo(object2);
							}
						}					
					});
					kItemsets.add(jointCandidate);
					//System.out.println("Candidate generated: " + Arrays.toString(jointCandidate.toArray()));
				}
			}
		}

		// Prune the candidates that do not meet the Apriori property, ie, subsets of a frequent itemset
		// must also frequent
		ArrayList<ArrayList<String>> prunedCandidates = new ArrayList<ArrayList<String>>();
		// only need to prune for k-item candidates of size >= 3
		if ((kMinus1 + 1) < 3) {
			prunedCandidates = kItemsets;
		}
		else {			
			itemsetIter = kItemsets.iterator();
			while (itemsetIter.hasNext()){
				ArrayList<String> itemset = itemsetIter.next();
				// generate (k-1)-items subsets from the k-items candidate itemsets			
				for (int i = 0; i < itemset.size(); i++) {
					ArrayList<String> kMinus1Subset = new ArrayList<String>();			
					for (int k = 0; k < itemset.size(); k++) {
						if (i != k) {
							kMinus1Subset.add(itemset.get(k));
						}
					}
					// check if this (k-1)-item subset is frequent			
					if (!isFrequentItemset(kMinus1Subset)) {
						// discard the k-items candidate
						//System.out.println("Candidate discarded: " + Arrays.toString(itemset.toArray()));
						//System.out.println("Candidate subset not frequent: " + Arrays.toString(kMinus1Subset.toArray()));
						break;
					}
					// found a frequent itemset
					prunedCandidates.add(itemset);					
				}
			}
		}	
		// output
		itemsetIter = prunedCandidates.iterator();
		while (itemsetIter.hasNext()){
			ArrayList<String> itemset = itemsetIter.next();
			String[] tempCandidate = new String[itemset.size()];
			itemset.toArray(tempCandidate);
			StringBuilder buf = new StringBuilder();
			for (String item : tempCandidate) {
				buf.append(item).append(' ');
			}
			try {
				context.write(new Text(buf.toString().trim()), new IntWritable(1));
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
}